#include <stdlib.h>
#include <string>
#include <iostream>
#include <memory>
#include <rise/common/exmacros.h>
#include <rise/common/Log.h>
#include <rise/common/Exception.h>
#include <rise/signal/SignalHandler.h>
#include <rise/threading/Thread.h>
#include <staff/common/Value.h>
#include <staff/common/QName.h>
#include <staff/common/Attribute.h>
#include <staff/common/DataObject.h>
#include <staff-mq/client/Message.h>
#include <staff-mq/client/Subscriber.h>

class SubscriberWorker: public smq::MessageHandler
{
public:
  virtual void operator()(smq::Subscriber& rSubscriber, const smq::Message& rMsg)
  {
    const staff::CDataObject& tdoMsg = rMsg.tdoMessage;

    std::cout << "\n\n\033[1mSubscriberWorker" << m_sName << ".\033[0m Message arrived: \n"
        << "msgid: " << rMsg.sId << std::endl
        << "queue: " << rMsg.sQueue << std::endl
        << "publisher: " << rMsg.sPublisher << std::endl
        << "Message: " << tdoMsg.GetText() << std::endl;
  }
  
  void Start(const std::string& sName, bool bStartInBackground, const std::string& sServerHost, unsigned short ushServerPort)
  {
    m_tSubscriber.Init("subscriber-" + sName, sServerHost, ushServerPort);

    std::cout << "subscribing to \"queue" << sName << "\"" << std::endl;
    m_tSubscriber.SetHandler(this);
    m_tSubscriber.Subscribe("queue" + sName);

    m_tSubscriber.Start(bStartInBackground);
    
    m_sName = sName;
  }
  
  void Wait()
  {
    m_tSubscriber.Wait();
  }

  void Stop()
  {
    m_tSubscriber.Stop();
  }
  
  virtual ~SubscriberWorker()
  {
  }

private:
  smq::Subscriber m_tSubscriber;
  std::string m_sName;
};


class MainApp: public rise::CSignalHandler
{
public:
  MainApp():
    m_bStop(false)
  {
  }

  void Start(const std::string& sServerHost, unsigned short ushServerPort)
  {
    SubscriberWorker tWorker1;
    SubscriberWorker tWorker2;
    
    tWorker1.Start("1", true, sServerHost, ushServerPort);
    tWorker2.Start("2", true, sServerHost, ushServerPort);
    
    m_hThreadMainId = rise::threading::CThread::GetCurrentId();
    RegisterSignal(ESIGINT);
    
    // main work cycle
    while (!m_bStop)
    {
      rise::threading::CThread::Sleep(1000);
    }
    
    tWorker1.Stop();
    tWorker2.Stop();
  }

protected:
  virtual void SignalHandler(int nSignal)
  {
    if (nSignal == ESIGINT)
    {
      if (rise::threading::CThread::GetCurrentId() == m_hThreadMainId)
      {
        m_bStop = true;
        rise::LogDebug() << "Breaking execution";
      } // child threads simply ignores this signal
    }
  }
  
private:
  bool m_bStop;
  rise::HThread m_hThreadMainId;
};


int main(int nArgs, char* szArgs[])
{
  std::string sServerHost = "127.0.0.1";
  unsigned short ushServerPort = 8768;

  for (int nArg = 1; nArg < nArgs; ++nArg)
  {
    if(szArgs[nArg])
    {
      const std::string& sArg = std::string(szArgs[nArg]);
      if (sArg.substr(0, 2) == "-s")
      {
        sServerHost = sArg.substr(2);
      }
      else
      if (sArg.substr(0, 2) == "-p")
      {
        ushServerPort = static_cast<unsigned short>(atoi(sArg.substr(2).c_str()));
      }
      else
      if (sArg.substr(0, 2) == "-h")
      {
        std::cerr << "Subscriber multithreading sample.\n\n"
            << szArgs[0] << "[ -h][ -s<Server host>][ -p<port>]\n\n";
        return 0;
      }
      else
      {
        std::cerr << "unknown arg: " << sArg << "\n";
        return 1;
      }
    }
  }
  
  try
  {
    MainApp tApp;
    tApp.Start(sServerHost, ushServerPort);
  }
  RISE_CATCH_ALL

  return 0;
}
